package com.netty.server.registry;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.netty.common.constant.Constant;


public class ServiceRegistry {
	
	private static final Logger logger = LoggerFactory.getLogger(ServiceRegistry.class);
	
	/**
	 * zk地址
	 */
	private String registryAddress;
	/**
	 * 计数器
	 */
    private CountDownLatch latch = new CountDownLatch(1);

	public ServiceRegistry(String registryAddress){
		this.registryAddress = registryAddress;
	}

	public void register(String serverAddress) {
		if (serverAddress != null) {
			ZooKeeper zk = connectServer();
			if (zk != null) {
				addRootNode(zk);
				createNode(zk,serverAddress);
			}
		}
		
	}
	
	
	/**
	 * 连接zk
	 * @return
	 */
	private ZooKeeper connectServer() {
		ZooKeeper zk = null;
		try {
			zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
				/**
				 * 回调通知成功后计数器减1
				 */
				@Override
				public void process(WatchedEvent event) {
					if (event.getState() == Event.KeeperState.SyncConnected) {
						latch.countDown();
					}
				}
			});
			latch.await();
		} catch (Exception e) {
			logger.error("", e);
		}
		return zk;
	}
	
	/**
	 * 添加根节点，根节点为持久节点
	 * @param zk
	 */
	private void addRootNode(ZooKeeper zk) {
		try {
			Stat s = zk.exists(Constant.ZK_REGISTRY_PATH, false);
			if (s == null) {
				/**
				 * 根节点为持久节点
				 */
				zk.create(Constant.ZK_REGISTRY_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		} catch (Exception e) {
			logger.error("",e);
		}
		
	}


	

	/**
	 * 创建数据节点, 顺序临时节点
	 * @param zk
	 * @param serverAddress
	 */
	private void createNode(ZooKeeper zk, String serverAddress) {
		try {
			byte[] bytes = serverAddress.getBytes();
			String path = zk.create(Constant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
			logger.debug("create zookeeper node ({} => {})", path, serverAddress);
		} catch (Exception e) {
			logger.error("",e);
		}
		
	}


}
